package com.ajun.producer;

import com.ajun.common.rabbit.QueueConstants;
import com.ajun.entity.Order;
import com.ajun.entity.OrderMessage;
import com.ajun.mapper.OrderMessageMapper;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * 发送订单 MQ 生产者
 *
 * @author LiJun
 * @date 2021-07-15
 */
@Slf4j
@Service
public class OrderProducer {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private OrderMessageMapper orderMessageMapper;

    @PostConstruct
    public void confirmCallBack() {
        // 消息发送成功后,给予生产者的消息回执,来确保生产者的可靠性
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            log.info("causer:{}", cause);
            // 发送消息时设置的相关数据
            String orderId = null;
            if (correlationData != null) {
                orderId = correlationData.getId();
            }

            // 查询该消息记录
            int orderIdNum = 0;
            if (orderId != null) {
                orderIdNum = Integer.parseInt(orderId);
            }
            OrderMessage orderMessage = orderMessageMapper.selectById(orderIdNum);

            // nack
            if (!ack) {
                // nack,改变消息状态为失败
                log.error("MQ队列应答失败,orderId:{}", orderId);
                if (orderMessage != null) {
                    orderMessage.setStatus((byte) -1);
                    int count = orderMessageMapper.updateById(orderMessage);
                    if (count == 1) {
                        log.error("本地消息状态修改,消息没有投递到消息队列中...");
                    }
                }
                return;
            }

            // ack
            try {
                if (orderMessage != null) {
                    orderMessage.setStatus((byte) 1);
                    int count = orderMessageMapper.updateById(orderMessage);
                    if (count == 1) {
                        log.info("本地消息状态修改成功,消息成功投递到消息队列中...");
                    }
                }
            } catch (Exception ex) {
                log.error("本地消息状态修改失败,出现异常:{}", ex.getMessage());
            }
        });
    }

    public void sendOrderMsg(Order order) {
        // 通过 MQ 发送消息
        rabbitTemplate.convertAndSend(
            QueueConstants.ORDER_DISPATCH_FANOUT_EXCHANGE,
            "",
            JSON.toJSONString(order),
            /*message -> {
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 持久化
                message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON); // JSON
                message.getMessageProperties().setContentEncoding("UTF_8");
                return message;
            },*/
            new CorrelationData(String.valueOf(order.getOrderId()))
        );
    }
}
